package nsqframe

import (
	"net"

	"github.com/nsqio/go-nsq"
	"github.com/xlvector/dlog"
)

type NFHandler interface {
	NFHandleMessage(msg []byte) error
}

//for api
type NFHandlerFunc func(msg []byte) error

func (h NFHandlerFunc) NFHandleMessage(msg []byte) error {
	return h(msg)
}

type NFConsumer struct {
	productCh chan []byte
	//exitCh    chan bool
	conf *NFCConfig

	nsqconsumer *nsq.Consumer
	handle      NFHandler

	handleError int64
	accept      int64
}

/*
  处理错误个数
*/
func (p *NFConsumer) HandleError() int64 {
	return p.handleError
}

func newNsqConsumer(nsqconf *NFNsqConfig) *nsq.Consumer {
	config := nsq.NewConfig()
	config.LocalAddr, _ = net.ResolveTCPAddr("tcp", ":0")
	c, err := nsq.NewConsumer(nsqconf.Topic, nsqconf.Chan, config)
	if err != nil {
		dlog.Fatal("new nsq consumer fail! %v", err)
	}

	return c
}

func NewNFConsumer(conf *NFCConfig, handle NFHandler) *NFConsumer {
	err := conf.check()
	if err != nil {
		dlog.Fatal("consumer config wrong!")
	}

	c := &NFConsumer{
		productCh: make(chan []byte, conf.CacheLen),
		//exitCh:      make(chan bool),
		conf:        conf,
		nsqconsumer: newNsqConsumer(conf.Nsq),
		handle:      handle,
	}

	c.nsqconsumer.AddHandler(c)
	err = c.nsqconsumer.ConnectToNSQD(conf.Nsq.Server)
	if err != nil {
		dlog.Fatal("connect nsq server fail! %v", err)
	}
	return c
}

func (p *NFConsumer) work(idx int) {
	dlog.Info("work: %d", idx)
	var product []byte
	var ok bool
	for {
		product, ok = <-p.productCh
		if !ok {
			break
		}
		err := p.handle.NFHandleMessage(product)
		if err != nil {
			p.handleError++
		}
	}

	dlog.Info("work exit! %d", idx)
}

func (p *NFConsumer) Start() {
	for i := 0; i < p.conf.Concurrent; i++ {
		go p.work(i)
	}
}

func (p *NFConsumer) Stop() {
	p.nsqconsumer.Stop()
	close(p.productCh)
}

func (p *NFConsumer) HandleMessage(msg *nsq.Message) error {
	p.accept++
	p.productCh <- msg.Body
	return nil
}
